package org.niit.service

import org.apache.spark.streaming.dstream.DStream
import org.niit.bean.AdClickData
import org.niit.common.TService
import org.niit.dao.AreaAdDao

import java.text.SimpleDateFormat
import java.util.Date

/**
 * Date:2025/6/4
 * Author：Ys
 * Description:
 */
class AreaCityAdService extends TService[DStream[AdClickData]]{

  override def dataAnalysis(data:DStream[AdClickData]): Unit = {


    //需求：统计每天 每地区 每城市 点击广告的次数
    //1. 第一步 将 AdClickData 封装的数据 转换成 （ （日期，区域，城市，广告），次数(1) ）
    val mapDS: DStream[((String, String, String, String), Int)] = data.map(line => {
      val area = line.area
      val city = line.city
      val ad = line.ad
      val sdf = new SimpleDateFormat("yyyy-MM-dd")
      val day = sdf.format(new Date(line.ts.toLong))

      ((day, area, city, ad), 1)

    })

    //2. 第二步，将数据进行聚合，将相同key的数据进行聚合，相同key的数据就是（ （日期，区域，城市，广告），次数(1) ）
    val reduceDS: DStream[((String, String, String, String), Int)] = mapDS.reduceByKey(_ + _)

    //3. 第三步,存入数据库
    reduceDS.foreachRDD(rdd=>{
      rdd.foreach{
        case ((day, area, city, ad), count) => {
          println(s"${day} ${area} ${city} ${ad} ${count}")
          //调用数据库的操作 写入数据库
          val areaAdDao = new AreaAdDao
          areaAdDao.insertAreaCityAd(day, area, city, ad, count)
        }
      }
    })

  }

}
